/*异步消息处理类*/
#pragma once
#include "buffer.hpp"
#include <atomic>
#include <memory>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>

namespace wu {
    using Func = std::function<void(Buffer &)>;
    enum class AsyncType {
        ASYNC_SAFE,//安全状态，生产者缓冲区满了就阻塞起来,避免资源耗尽风险
        ASYNC_UNSAFE//不安全状态，不考虑资源问题，无线扩容，用于测试
    };
    class AsyncLooper {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Func &funt, AsyncType looper_type = AsyncType::ASYNC_SAFE)
        :_stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)), _callback(funt),_looper_type(looper_type)
        {}
        ~AsyncLooper() { stop(); }
        void stop() {
            _stop = true;
            //异步日志消息处理器停止之前要将所有线程的工作完成之后再退出
            _cond_con.notify_all();
            _thread.join();
        }
        void push(const char* data, size_t len) {
            //生产者缓冲区有两种方案1.写满就阻塞  2.无线扩容(用于日志器的性能测试)
            std::unique_lock<std::mutex> lock(_mutex);
            //如果缓冲区的剩余空间大小大于len就可以写。
            if(_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&](){return _pro_buf.writeOfSize() >= len; });
            //向缓冲区写入
            _pro_buf.push(data, len);
            //唤醒消费者对消费者缓冲区进行读取
            _cond_con.notify_one();
        }
    private:
        void threadEntry() {
            while(1) {
                //为互斥锁设置一个生命周期,当缓冲区交换完之后就解锁，并不等待数据处理的过程，提高异步的性能
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    //当退出标志位被设置，并且缓冲区中已经没有数据了再退出，非则会导致程序退出了，但是缓冲区的数据还没处理完
                    if(_stop && _pro_buf.empty()) break;
                    
                    //1.判断当前生产者缓冲区有没有数据，有则交换，没有则阻塞等待
                    _cond_con.wait(lock, [&](){return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    //2.唤醒生产者向生产者缓冲区进行写入
                    if(_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                //3.对缓冲区的数据进行处理
                _callback(_con_buf);
                //4.初始化消费缓冲区
                _con_buf.reset();
            }
        }
    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop;
        Buffer _pro_buf;//生产缓冲区
        Buffer _con_buf;//消费缓冲区
        std::mutex _mutex;//
        std::condition_variable _cond_pro;//生产者
        std::condition_variable _cond_con;//消费者
        std::thread _thread;
        Func _callback;//回调函数
    };
}


